package com.vip.influx.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.influxdb.InfluxDB;
import org.influxdb.dto.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * influx数据库工具类
 *
 * @author wgb
 * @date 2020/6/16 15:42
 */
@Component
public class InfluxDBUtils {

    @Autowired
    private InfluxDB influxDB;

    /**
     * 插入数据
     *
     * @param database        数据库名
     * @param measurement     表名
     * @param tags            标签
     * @param fields          存储对象
     * @param timestamp       时间戳
     * @param retentionPolicy 存储策略
     */
    public void insert(String database, String measurement, Map<String, String> tags, Object fields, long timestamp, String retentionPolicy) {
        Point.Builder builder = Point.measurement(measurement);
        builder.tag(tags);
        builder.fields(JSONObject.parseObject(JSON.toJSONString(fields)));
        if (0 != timestamp) {
            builder.time(timestamp + 28800000, TimeUnit.MILLISECONDS);
        }
        influxDB.write(database, retentionPolicy, builder.build());
    }

    /**
     * 插入数据
     *
     * @param database        数据库名
     * @param measurement     表名
     * @param fields          存储对象
     * @param timestamp       时间戳
     * @param retentionPolicy 存储策略
     */
    public void insert(String database, String measurement, Object fields, long timestamp, String retentionPolicy) {
        Point.Builder builder = Point.measurement(measurement);
        builder.fields(JSONObject.parseObject(JSON.toJSONString(fields)));
        if (0 != timestamp) {
            builder.time(timestamp + 28800000, TimeUnit.MILLISECONDS);
        }
        influxDB.write(database, retentionPolicy, builder.build());
    }

    /**
     * 插入数据
     *
     * @param database        数据库名
     * @param measurement     表名
     * @param fields          存储对象
     * @param retentionPolicy 存储策略
     */
    public void insert(String database, String measurement, Object fields, String retentionPolicy) {
        Point.Builder builder = Point.measurement(measurement);
        builder.fields(JSONObject.parseObject(JSON.toJSONString(fields)));
        builder.time(System.currentTimeMillis() + 28800000, TimeUnit.MILLISECONDS);
        influxDB.write(database, retentionPolicy, builder.build());
    }

    /**
     * 插入数据
     *
     * @param database    数据库名
     * @param measurement 表名
     * @param fields      存储对象
     */
    public void insert(String database, String measurement, Object fields) {
        Point point = Point.measurement(measurement).fields(JSONObject.parseObject(JSON.toJSONString(fields)))
              .time(System.currentTimeMillis() + 28800000, TimeUnit.MILLISECONDS).build();
        influxDB.setDatabase(database).write(point);
    }


    /**
     * 为指定数据库创建自定义保留策略
     *
     * @param policyName  策略名
     * @param database    数据库名
     * @param duration    保存天数
     * @param replication 保存副本数量
     * @param isDefault   是否设为默认保留策略
     */
    public void createRetentionPolicy(String policyName, String database, String duration, int replication, boolean isDefault) {
        String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s", policyName, database, duration, replication);
        if (isDefault) {
            sql = sql + " DEFAULT";
        }
        this.query(sql, database);
    }

    /**
     * 查询
     *
     * @param command 查询语句
     * @return
     */
    public QueryResult query(String command, String database) {
        return influxDB.query(new Query(command, database));
    }

    /**
     * 创建数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void createDB(String dbName) {
        influxDB.createDatabase(dbName);
    }

    /**
     * 删除数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void deleteDB(String dbName) {
        influxDB.deleteDatabase(dbName);
    }

    /**
     * 测试连接是否正常
     *
     * @return true 正常
     */
    public boolean ping() {
        boolean isConnected = false;
        try {
            Pong pong = influxDB.ping();
            if (pong != null) {
                isConnected = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isConnected;
    }

    /**
     * 删除
     *
     * @param command  删除语句
     * @param database 数据库
     * @return 返回错误信息
     */
    public String deleteMeasurementData(String command, String database) {
        QueryResult result = influxDB.query(new Query(command, database));
        return result.getError();
    }

    /**
     * 批量写入测点
     *
     * @param batchPoints
     */
    public void insertMulti(BatchPoints batchPoints) {
        influxDB.write(batchPoints);
    }

    /**
     * 批量写入数据
     *
     * @param database        数据库
     * @param retentionPolicy 保存策略
     * @param consistency     一致性
     * @param records         要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）
     */
    public void insertMulti(final String database, final String retentionPolicy, final InfluxDB.ConsistencyLevel consistency, final List<String> records) {
        influxDB.write(database, retentionPolicy, consistency, records);
    }

}
